package org.atlantis.datasink.event;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public final class EventManager {

	public static final int DEFAULT_MESSAGE_LIMIT = 512;
	public static final int DEFAULT_WORKER_LIMIT = 3;

	private static class Holder {
		static EventManager mgr = new EventManager();
	}

	private ExecutorService executor = Executors.newCachedThreadPool(new EventNamedThreadFactory());
	private ConcurrentMap<String, EventQueue> eventQueues = new ConcurrentHashMap<String, EventQueue>();

	private volatile boolean running = true;

	public static EventManager getInstance() {
		return Holder.mgr;
	}

	private EventManager() {}

	public EventQueue get(String name) {
		assertAlive();
		EventQueue queue = eventQueues.get(name);
		if (queue == null) return null;
		if (!queue.isAlive()) {
			queue = null;
			eventQueues.remove(name);
		}
		return queue;
	}

	public EventQueue create(String name) {
		return create(name, DEFAULT_MESSAGE_LIMIT, DEFAULT_WORKER_LIMIT);
	}

	public EventQueue create(String name, int messageLimit, int workerLimit) {
		assertAlive();
		if (eventQueues.containsKey(name)) throw new IllegalArgumentException("Duplicate event name: " + name);
		EventQueue queue = new LimitedEventQueue(name, messageLimit, workerLimit);
		EventQueue oldQueue = eventQueues.putIfAbsent(name, queue);
		if (oldQueue != null) {
			queue.stop();
			throw new IllegalArgumentException("Duplicate event name: " + name);
		}
		return queue;
	}

	public EventQueue getOrCreate(String name) {
		return getOrCreate(name, DEFAULT_MESSAGE_LIMIT, DEFAULT_WORKER_LIMIT);
	}

	public EventQueue getOrCreate(String name, int messageLimit, int workerLimit) {
		EventQueue queue = get(name);
		if (queue != null) return queue;
		queue = new LimitedEventQueue(name, messageLimit, workerLimit);
		EventQueue oldQueue = eventQueues.putIfAbsent(name, queue);
		if (oldQueue != null) {
			queue.stop();
			return oldQueue;
		}
		return queue;
	}

	public Map<String, Collection<Object>> stop() {
		running = false;

		Map<String, Collection<Object>> remainingMessages = new HashMap<String, Collection<Object>>();

		for (EventQueue queue : eventQueues.values())
			remainingMessages.put(queue.getName(), queue.stop());
		eventQueues.clear();
		eventQueues = null;
		executor.shutdownNow();
		return remainingMessages;
	}

	private void assertAlive() {
		if (!running) throw new IllegalStateException("EventManager is stoped.");
	}

	void execute(Runnable task) {
		executor.execute(task);
	}
}
